package hbasedeploylogic

import (
	"context"
	"errors"
	"fmt"
	"github.com/zeromicro/go-zero/core/logx"
	"strings"
	"sync"
	"time"
	"yunzhan/common/models"
	utils "yunzhan/common/utils"
	"yunzhan/rpc-server/internal/svc"
	agent "yunzhan/rpc-server/pb"
)

type DeployHBaseHALogic struct {
	ctx    context.Context
	svcCtx *svc.ServiceContext
	logx.Logger
}

func NewDeployHBaseHALogic(ctx context.Context, svcCtx *svc.ServiceContext) *DeployHBaseHALogic {
	return &DeployHBaseHALogic{
		ctx:    ctx,
		svcCtx: svcCtx,
		Logger: logx.WithContext(ctx),
	}
}

// DeployHBaseHA 部署HBase集群
func (l *DeployHBaseHALogic) DeployHBaseHA(in *agent.DeployHBaseRequest) (*agent.DeployResponse, error) {
	if in.NodeInfo == nil {
		return &agent.DeployResponse{
			Code:    500,
			Message: "未配置节点信息, 无法执行HBase集群部署",
		}, errors.New("未配置节点信息, 无法执行HBase集群部署")
	}

	// 立即返回一个响应，表示请求已接收并正在处理
	resp := &agent.DeployResponse{
		Code:    200,
		Message: fmt.Sprintf("请求已接收，正在部署 %s 到节点 %s", in.ConfigInfo.ComponentName, in.NodeInfo.Host),
	}

	go l.startDeployment(in)

	return resp, nil
}

func (l *DeployHBaseHALogic) startDeployment(in *agent.DeployHBaseRequest) {

	var wg sync.WaitGroup

	wg.Add(1)
	go func() {
		defer wg.Done()

		// 创建 LogManager 实例
		logManager := models.NewLogManager(100, fmt.Sprintf("http://%s:%d", l.svcCtx.Config.RestServices["apiServer"].RestConf.Host, l.svcCtx.Config.RestServices["apiServer"].RestConf.Port))
		defer logManager.Close()

		flushInterval := 5 * time.Second // 刷新间隔
		maxBatchSize := 100              // 最大批量大小
		retryDelay := 2 * time.Second    // 重试延迟
		maxRetries := 3                  // 最大重试次数

		// 创建 ComponentInfoManager 实例
		componentInfoManager := models.NewComponentInfoManager(
			1000,
			fmt.Sprintf("http://%s:%d", l.svcCtx.Config.RestServices["apiServer"].RestConf.Host, l.svcCtx.Config.RestServices["apiServer"].RestConf.Port),
			flushInterval,
			maxBatchSize,
			retryDelay,
			maxRetries)
		defer componentInfoManager.Close()
		packageDirectory := in.ConfigInfo.PackagePath
		rtrimDir := utils.Rtrim(packageDirectory, "/")
		hbaseVersion := in.ConfigInfo.ComponentName
		installPath := utils.Rtrim(in.ConfigInfo.InstallPath, "/")
		baseInfo := fmt.Sprintf("安装包存放目录: %s, HBase version: %s, HBase 安装目录: %s", rtrimDir, hbaseVersion, installPath)
		step := 1
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", baseInfo, "HBase", "Server")

		javaHome := "/usr/local/java"

		// 检测安装路径是否存在
		checkPathCommand := fmt.Sprintf(`
if [ ! -d "%s" ]; then
   sudo mkdir -p "%s"
fi
`, installPath, installPath)

		step++
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "执行组件安装路径检查...", "HBase", "Server")
		output, err := utils.ExecCommand(checkPathCommand)
		if err != nil {
			message := fmt.Sprintf("检查组件安装路径异常: %v, output: %s", err, output)
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "ERROR", message, "HBase", "Server")
			return
		}
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "检查组件安装路径完成", "HBase", "Server")

		step++
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "解压安装包...", "HBase", "Server")
		tarCommand := fmt.Sprintf("sudo tar -xzf %s/%s -C %s", packageDirectory, hbaseVersion, installPath)
		output, err = utils.ExecCommand(tarCommand)
		if err != nil {
			message := fmt.Sprintf("解压安装包异常: %v, output: %s", err, output)
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "ERROR", message, "HBase", "Server")
			return
		}
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "解压安装包完成", "HBase", "Server")

		hdfsHome := utils.Rtrim(hbaseVersion, ".tar.gz")
		home := fmt.Sprintf("%s/%s", installPath, hdfsHome)

		step++
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "修改组件所属用户组...", "HBase", "Server")
		chownCommand := fmt.Sprintf("sudo chown -R %s:%s %s", in.NodeInfo.Username, in.NodeInfo.Username, home)
		output, err = utils.ExecCommand(chownCommand)
		if err != nil {
			message := fmt.Sprintf("修改组件所属用户组异常: %v, output: %s", err, output)
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "ERROR", message, "HBase", "Server")
			return
		}
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "修改组件所属用户组完成", "HBase", "Server")

		checkPathCommand = fmt.Sprintf(`
if [ ! -d "%s" ]; then
   sudo mkdir -p "%s"
fi
sudo chown -R %s:%s %s
`, in.HbaseConfig.HbaseLocalDir,
			in.HbaseConfig.HbaseLocalDir,
			in.NodeInfo.Username,
			in.NodeInfo.Username,
			in.HbaseConfig.HbaseLocalDir)
		step++
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "检查组件数据存储目录...", "HBase", "Server")
		output, err = utils.ExecCommand(checkPathCommand)
		if err != nil {
			message := fmt.Sprintf("检查组件数据存储目录异常: %v, output: %s", err, output)
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "ERROR", message, "HBase", "Server")
			return
		}
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "检查组件数据存储目录完成", "HBase", "Server")

		step++
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "加载hbase-site.xml配置文件...", "HBase", "Server")
		hbaseSitePath := fmt.Sprintf("%s/conf/hbase-site.xml", hdfsHome)
		hbaseConfig, err := utils.LoadConfiguration(hbaseSitePath)
		if err != nil {
			message := fmt.Sprintf("加载hbase-site.xml配置文件异常: %v, output: %s", err, output)
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "ERROR", message, "HBase", "Server")
			return
		}
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "加载 hbase-site.xml 配置文件完成", "HBase", "Server")

		hbaseConfig.SetProperty("hbase.master.fileSplitTimeout", in.HbaseConfig.HbaseMasterFileSplitTimeout, "拆分一个region时，在文件拆分步骤中等待的时间（毫秒），超过此时间将中止尝试。默认值：600000。")
		hbaseConfig.SetProperty("hbase.regionserver.port", in.HbaseConfig.HbaseRegionServerPort, "HBase RegionServer 绑定的端口。")
		hbaseConfig.SetProperty("hbase.regionserver.info.port", in.HbaseConfig.HbaseRegionServerInfoPort, "HBase RegionServer Web UI 的端口。设置为-1如果不希望运行RegionServer UI。")
		hbaseConfig.SetProperty("hbase.regionserver.handler.count", in.HbaseConfig.HbaseRegionServerHandlerCount, "在RegionServers上启动的RPC监听器实例的数量。太多的处理程序可能适得其反。使其成为CPU数量的倍数。如果主要是只读，处理程序数量接近CPU数量效果更好。从CPU数量的两倍开始进行调整。")
		hbaseConfig.SetProperty("hbase.master.logcleaner.ttl", in.HbaseConfig.HbaseMasterLogcleanerTTL, "一个WAL在存档目录（{hbase.rootdir}/oldWALs）中保留的时间（毫秒），之后将由Master线程清理。")
		hbaseConfig.SetProperty("hbase.master.logcleaner.plugins", in.HbaseConfig.HbaseMasterLogcleanerPlugins, "由LogsCleaner服务调用的BaseLogCleanerDelegate的逗号分隔列表。")
		hbaseConfig.SetProperty("hbase.master.port", in.HbaseConfig.HbaseMasterPort, "HBase Master 绑定的端口。")
		hbaseConfig.SetProperty("hbase.master.info.port", in.HbaseConfig.HbaseMasterInfoPort, "HBase Master Web UI 的端口。设置为-1如果不希望运行UI实例。")
		hbaseConfig.SetProperty("hbase.local.dir", in.HbaseConfig.HbaseLocalDir, "在本地文件系统上用作本地存储的目录。")
		hbaseConfig.SetProperty("hbase.client.write.buffer", in.HbaseConfig.HbaseClientWriteBuffer, "BufferedMutator写缓冲区的默认大小（字节）。较大的缓冲区会占用更多内存，但较大的缓冲区大小会减少发起的RPC数量。")
		hbaseConfig.SetProperty("hbase.client.max.total.tasks", in.HbaseConfig.HbaseClientMaxTotalTasks, "单个HTable实例将发送到集群的最大并发变更任务数。")
		hbaseConfig.SetProperty("hbase.normalizer.merge.min.region.count", in.HbaseConfig.HbaseNormalizerMergeMinRegionCount, "考虑合并规范化的表的最小region数量。")
		hbaseConfig.SetProperty("hbase.normalizer.merge.min_region_age.days", in.HbaseConfig.HbaseNormalizerMergeMinRegionAgeDays, "将region视为合并的最小年龄，以天为单位。")
		hbaseConfig.SetProperty("hbase.normalizer.merge.min_region_size.mb", in.HbaseConfig.HbaseNormalizerMergeMinRegionSizeMb, "将region视为合并的最小大小，以MB为单位。")
		hbaseConfig.SetProperty("hbase.hregion.memstore.flush.size", in.HbaseConfig.HbaseHRegionMemStoreFlushSize, "如果memstore的大小超过此字节数，将刷写到磁盘。值由每个hbase.server.thread.wakefrequency运行的线程检查。")
		hbaseConfig.SetProperty("hbase.hregion.max.filesize", in.HbaseConfig.HbaseHRegionMaxFilesize, "最大文件大小。如果region的HFiles的总大小超过此值，region将被拆分为两个。")
		hbaseConfig.SetProperty("hbase.rest.port", in.HbaseConfig.HbaseRestPort, "HBase REST服务器的端口。")
		hbaseConfig.SetProperty("hbase.client.scanner.caching", in.HbaseConfig.HbaseClientScannerCaching, "扫描的缓存行数，值越大，消耗的内存越大")

		step++
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "更新hbase-site.xml配置文件...", "HBase", "Server")
		err = hbaseConfig.Save(hbaseSitePath)
		if err != nil {
			message := fmt.Sprintf("更新 hbase-site.xml 配置文件异常: %v, output: %s", err, output)
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "ERROR", message, "HBase", "Server")
			return
		}
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "更新hbase-site.xml配置文件完成", "HBase", "Server")

		step++
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "配置hbase-env.sh文件...", "HBase", "Server")
		hadoopEnvPath := fmt.Sprintf("%s/conf/hbase-env.sh", hdfsHome)
		appendHadoopEnvCommand := fmt.Sprintf(`
sudo bash -c 'cat <<EOF >>%s
export JAVA_HOME=%s
export YARN_RESOURCEMANAGER_OPTS="-XX:+AlwaysPreTouch -Xss1m  -Xms2g -Xmx2g -XX:+UseG1GC -XX:+HeapDumpOnOutOfMemoryError -XX:G1HeapRegionSize=4m -XX:InitiatingHeapOccupancyPercent=30 -XX:G1ReservePercent=15"
EOF'
`, hadoopEnvPath, javaHome)
		output, err = utils.ExecCommand(appendHadoopEnvCommand)
		if err != nil {
			message := fmt.Sprintf("配置 hbase-env.sh 文件异常: %v, output: %s", err, output)
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "ERROR", message, "HBase", "Server")
			return
		}
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "配置hbase-env.sh文件完成", "HBase", "Server")

		step++
		currentIP, currentIPErr := utils.GetCurrentInternalIP()
		if currentIPErr != nil {
			message := fmt.Sprintf("无法获取当前主机IP: %v", currentIPErr)
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "ERROR", message, "HBase", "Server")
			return
		}
		currentHostname, currentHostErr := utils.GetCurrentHostname()
		if currentHostErr != nil {
			message := fmt.Sprintf("无法获取当前主机名: %v", currentHostErr)
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "ERROR", message, "HBase", "Server")
			return
		}

		// HMaster部署
		if utils.Contains(in.HbaseConfig.MasterServers, currentHostname) || utils.Contains(in.HbaseConfig.MasterServers, currentIP) {

			masterServiceFile := fmt.Sprintf(`[Unit]
Description=HBase HMaster Service
After=network.target
Wants=network-online.target
After=network-online.target

[Service]
User=%s
Group=%s
Environment="JAVA_HOME=%s"
Environment="HBASE_HOME=%s"
ExecStart=%s/bin/hbase-daemon.sh start master
ExecStop=%s/bin/hbase-daemon.sh stop  master

Restart=on-failure
RestartSec=10
KillSignal=SIGINT
SyslogIdentifier=hbase-master
LimitNOFILE=65536

[Install]
WantedBy=multi-user.target`, in.NodeInfo.Username, in.NodeInfo.Username, javaHome, home, home, home)
			step++
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "创建 hbase-master.service 文件...", "HBase", "HMaster")
			writeServiceFileCommand := fmt.Sprintf("echo '%s' | sudo tee /etc/systemd/system/hbase-master.service", masterServiceFile)
			_, err = utils.ExecCommand(writeServiceFileCommand)
			if err != nil {
				message := fmt.Sprintf("创建 hbase-master.service 文件异常: %v, output: %s", err, output)
				logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "ERROR", message, "HBase", "HMaster")
				return
			}
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "创建 hbase-master.service 完成", "HBase", "HMaster")

			step++
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "启动HMaster服务...", "HBase", "HMaster")
			journalNodeStartCommand := fmt.Sprintf("sudo systemctl daemon-reload && sudo systemctl enable hbase-master.service --now")
			output, err = utils.ExecCommand(journalNodeStartCommand)
			if err != nil {
				message := fmt.Sprintf("启动HMaster服务异常: %v, output: %s", err, output)
				logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "ERROR", message, "HBase", "HMaster")
				return
			}
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "启动HMaster服务完成", "HBase", "HMaster")
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "等待HMaster服务完全启动", "HBase", "HMaster")

			time.Sleep(3 * time.Second)
			step++
			statusCommand := "sudo systemctl is-active hbase-master.service"
			output, err = utils.ExecCommand(statusCommand)
			if err != nil || strings.TrimSpace(output) != "active" {
				message := fmt.Sprintf("HMaster服务启动后未正常运行: %v, output: %s", err, output)
				logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "ERROR", message, "HBase", "HMaster")
				return
			}
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "HMaster部署成功", "HBase", "HMaster")

			hbaseInfo := &models.ComponentInfo{
				ClusterID:      in.ClusterID,
				ComponentName:  "HBase",
				Version:        in.Version,
				NodeHost:       currentHostname,
				NodeIP:         currentIP,
				ComponentRole:  "HMaster",
				HomePath:       home,
				DataStorageDir: in.HbaseConfig.HbaseLocalDir,
				Port:           in.HbaseConfig.HbaseMasterPort,
				Status:         true,
				AdditionalInfo: fmt.Sprintf("config_file=%s/conf/hbase-site.xml", home),
			}

			componentInfoManager.SendComponentInfo(hbaseInfo)

		}

		// HRegionServer部署
		if utils.Contains(in.HbaseConfig.RegionServers, currentHostname) || utils.Contains(in.HbaseConfig.RegionServers, currentIP) {

			regionServiceFile := fmt.Sprintf(`[Unit]
Description=HBase RegionServer Service
After=network.target
Wants=network-online.target
After=network-online.target

[Service]
User=%s
Group=%s
Environment="JAVA_HOME=%s"
Environment="HBASE_HOME=%s"
ExecStart=%s/bin/hbase-daemon.sh start regionserver
ExecStop=%s/bin/hbase-daemon.sh stop regionserver
Restart=on-failure
RestartSec=10
KillSignal=SIGINT
SyslogIdentifier=hbase-regionserver
LimitNOFILE=65536

[Install]
WantedBy=multi-user.target`, in.NodeInfo.Username, in.NodeInfo.Username, javaHome, home, home)

			step++
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "创建hbase-regionserver.service文件...", "HBase", "HRegionServer")
			writeServiceFileCommand := fmt.Sprintf("echo '%s' | sudo tee /etc/systemd/system/hbase-regionserver.service", regionServiceFile)
			_, err = utils.ExecCommand(writeServiceFileCommand)
			if err != nil {
				message := fmt.Sprintf("创建 hbase-regionserver.service文件异常: %v, output: %s", err, output)
				logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "ERROR", message, "HBase", "HRegionServer")
				return
			}
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "创建hbase-regionserver.service文件完成", "HBase", "HRegionServer")

			step++
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "启动HRegionServer服务...", "HBase", "Server")
			journalNodeStartCommand := fmt.Sprintf("sudo systemctl daemon-reload && sudo systemctl enable hbase-regionserver.service --now")
			output, err = utils.ExecCommand(journalNodeStartCommand)
			if err != nil {
				message := fmt.Sprintf("启动HRegionServer服务异常: %v, output: %s", err, output)
				logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "ERROR", message, "HBase", "HRegionServer")
				return
			}
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "启动HRegionServer服务完成", "HBase", "HRegionServer")
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "等待HRegionServer服务完全启动", "HBase", "HRegionServer")

			time.Sleep(3 * time.Second)
			step++
			statusCommand := "sudo systemctl is-active hbase-regionserver.service"
			output, err = utils.ExecCommand(statusCommand)
			if err != nil || strings.TrimSpace(output) != "active" {
				message := fmt.Sprintf("HRegionServer服务启动后未正常运行: %v, output: %s", err, output)
				logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "ERROR", message, "HBase", "HRegionServer")
				return
			}
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "HRegionServer部署成功", "HBase", "HRegionServer")

			hbaseInfo := &models.ComponentInfo{
				ClusterID:      in.ClusterID,
				ComponentName:  "HBase",
				Version:        in.Version,
				NodeHost:       currentHostname,
				NodeIP:         currentIP,
				ComponentRole:  "HRegionServer",
				HomePath:       home,
				DataStorageDir: in.HbaseConfig.HbaseLocalDir,
				Port:           in.HbaseConfig.HbaseRegionServerPort,
				Status:         true,
				AdditionalInfo: fmt.Sprintf("config_file=%s/conf/hbase-site.xml", home),
			}
			componentInfoManager.SendComponentInfo(hbaseInfo)
		}
	}()

	wg.Wait()

}
